package mutli_cache

import (
	"fmt"
	jsoniter "github.com/json-iterator/go"
	"github.com/mohae/deepcopy"
	"sync"
	"sync/atomic"
	"time"
)

// The value can be changed by external assignment
// It does not need to be too large, only one area needs to be locked, and it is very suitable if there are not many concurrent keys.
const (
	defaultShards         = 256
	defaultShardsAndOpVal = 255
)

var (
	CacheDebug                  bool   // 是否开启Debug模式,debug日志输出
	CacheMonitor                bool   // 缓存监视器,负责监视缓存是否过期
	OutCachePubSub              bool   // 缓存数据源,发布订阅模式是否开启
	CacheMonitorSecond          = 5    // 监视器默认扫描时间5秒
	DefaultGPoolWorkerNum       = 200  // 默认工作协程数目
	DefaultGPoolJobQueueChanLen = 1000 // 默认工作协程任务数量
)

var HitStatisticsOut HitStatistics // 统计变量
var json = jsoniter.ConfigCompatibleWithStandardLibrary

type MultiCache struct {
	CacheID    string                    // 缓存唯一GID
	out        OutCache                  // 缓存数据源
	local      LocalCache                // 本地缓存
	shardsLock [defaultShards]sync.Mutex // 分片锁
	hash       Harsher                   // 哈希
	isStop     chan struct{}             // 缓存停止channel,用于接收缓存关闭状态
	stopLock   sync.Once                 // 缓存关闭锁
	channel    chan *ChannelMeta         // 数据channel
	workerPool *WorkerPool               // 线程池
}

func New(out OutCache, local LocalCache) (g *MultiCache, err error) {
	// 如果传入本地缓存为空,默认新建一个本地缓存
	if local == nil {
		local = NewFreeCache()
	}
	// 传入外部缓存为空,新建一个redis外部缓存
	if out == nil {
		out, err = NewRedisCache()
		if err != nil {
			return nil, fmt.Errorf("获取NewRedisCache err: %v", err)
		}
	}
	// 获取当前cache CacheID
	gid, err := NewUUID()
	if err != nil {
		return nil, fmt.Errorf("获取MultiCache.gid err :%v", err)
	}
	LogInfo("[cache.gid] = ", gid)
	g = &MultiCache{
		CacheID: gid,
		// 哈希对象,用于获取传入对象哈希值
		hash:   new(fnv64a),
		isStop: make(chan struct{}, 1),
		// 256个缓存长度的channel
		channel: make(chan *ChannelMeta, defaultShards),
		// 创建线程池用于执行任务
		workerPool: NewPool(DefaultGPoolWorkerNum, DefaultGPoolJobQueueChanLen),
		local:      local,
		out:        out,
	}
	// 如果外部缓存实现了发布订阅接口,且打开了发布订阅模式
	_, ok := g.out.(PubSub)
	if ok && OutCachePubSub {
		// 使用传入缓存数据源的发布订阅接口
		g.workerPool.SendJob(wrapFuncErr(g.subscribe))
	}

	HitStatisticsOut.AccessGetTotal = 1
	// 如果打开了缓存监视器
	if CacheMonitor {
		// 将监视器函数传入工作队列
		g.workerPool.SendJob(g.monitor)
	}

	return g, nil
}

func (g *MultiCache) monitor() {
	// 定时器.每5秒执行一次,统计订阅数据
	t := time.NewTicker(time.Duration(CacheMonitorSecond) * time.Second)
	for {
		select {
		case <-g.isStop:
			return
		case <-t.C:
			HitStatisticsOut.Calculation()
			LogDebugF("statistics [\u001B[32m本地数据源\u001B[0m]命中率 = [%.4f]", HitStatisticsOut.HitLocalStorageTotalRate*100)
			LogDebugF("statistics [\u001B[33m缓存数据源\u001B[0m]命中率 = [%.4f]", HitStatisticsOut.HitOutStorageTotalRate*100)
			LogDebugF("statistics [\u001B[31m外部数据源\u001B[0m]命中率 = [%.4f]", HitStatisticsOut.HitDataSourceTotalRate*100)
		}
	}
}

// Get 获取数据
// key 数据对应key
// ttlSecond 超时时间
// obj 赋值对象,外部传入obj,obj会被深拷贝为对于的数据
// fn 加载数据的函数
func (g *MultiCache) Get(key string, ttlSecond int, obj interface{}, fn LoadDataSourceFunc) error {
	select {
	case <-g.isStop:
		// 缓存关闭
		return CacheClose
	default:
	}
	// 校验数据不为空
	if key == "" {
		return CacheKeyEmpty
	}
	// 校验传入数据部位nil,避免赋值失败
	if obj == nil {
		return CacheObjNil
	}
	// 校验获取数据函数不为空
	if fn == nil {
		return LoadDataSourceFuncNil
	}
	// 超时时间为0的话默认赋值5秒
	if ttlSecond <= 0 {
		ttlSecond = 5
	}
	return g.get(key, ttlSecond, obj, fn)
}

func (g *MultiCache) get(key string, ttlSecond int, obj interface{}, fn LoadDataSourceFunc) error {
	// 获取数据次数+1
	atomic.AddInt64(&HitStatisticsOut.AccessGetTotal, 1)
	// 首先从本地缓存中获取数据,本地数据源不需要深拷贝
	v, ok, err := g.local.Get(key, obj)
	if err != nil {
		return err
	}
	// 如果本地数据源获取成功
	if ok {
		// 本地数据命中次数+1
		atomic.AddInt64(&HitStatisticsOut.HitLocalStorageTotal, 1)
		if CacheDebug {
			LogDebugF("key:%-15s => [\u001B[32m hit local storage \u001B[0m]\n", key)
		}
		// 校验数据是否需要更新
		if v.Obsoleted() {
			// 如果数据需要更新,深拷贝一份数据
			to := deepcopy.Copy(obj) // async so copy obj
			g.workerPool.SendJob(func() {
				// 将数据同步到本地缓存中
				err := g.syncLocalCache(key, ttlSecond, to, fn)
				if err != nil {
					LogErrF("同步缓存数据 key=%s,err=%v\n", key, err)
				}
			})
		}
		// 将获取的数据v存入obj并返回
		return clone(v.Value, obj)
	}
	// 如果本地不存在数据,获取缓存数据源
	v, ok, err = g.out.Get(key, obj)
	if err != nil {
		return err
	}
	// 缓存数据源获取成功
	if ok {
		// 检查数据是否过期
		if !v.Expired() {
			// 外部数据命中次数+1
			atomic.AddInt64(&HitStatisticsOut.HitOutStorageTotal, 1)
			if CacheDebug {
				LogDebugF("key:%-15s => [\u001B[33m hit out storage \u001B[0m]\n", key)
			}
			// 将数据更新到本地
			err = g.local.Set(key, v)
			if err != nil {
				return err
			}
			// 深拷贝返回数据
			return clone(v.Value, obj)
		}
	}
	// 如果本地或缓存数据源均没有数据
	if fn != nil {
		// 使用外部数据源获取数据,并同步到本地以及缓存缓存中
		o, err := g.syncOutCache(key, ttlSecond, fn)
		if err != nil {
			return err
		}
		// 深拷贝返回
		return clone(o, obj)
	}
	// 所有数据源都不存在数据,返回错误
	return OutStorageLoadNil
}

func (g *MultiCache) syncLocalCache(key string, ttlSecond int, obj interface{}, fn LoadDataSourceFunc) error {
	// 因为数据失效,先从缓存中数据源获取实时数据
	e, ok, err := g.out.Get(key, obj)
	if err != nil {
		return err
	}
	// 如果外部数据不存在,获取外部缓存数据过期
	if !ok || e.Expired() {
		atomic.AddInt64(&HitStatisticsOut.HitDataSourceTotal, 1)
		if CacheDebug {
			LogDebugF("key:%-15s => [\u001B[31m hit data source \u001B[0m]\n", key)
		}
		// 先上一把分片锁
		g.getShardsLock(key)
		defer g.releaseShardsLock(key)
		// 从传入获取数据函数中获取数据
		v, err := fn()
		if err != nil {
			return err
		}
		// 数据不存在返回nil
		if v == nil {
			return DataSourceLoadNil
		}
		// 获得数据则构造entry
		e = NewEntry(v, ttlSecond)
		// 将entry存入本地缓存
		err = g.local.Set(key, e)
		if err != nil {
			return err
		}
		// 将entry存入缓存数据源
		g.workerPool.SendJob(func() {
			_err := g.out.Set(key, e)
			if _err != nil {
				LogErr("syncLocalCache out set err:", err)
				return
			}
			// 如果缓存数据源实现了发布订阅接口,还需要通知到对应的集群
			pubsub, ok := g.out.(PubSub)
			if ok && OutCachePubSub {
				_err = pubsub.Publish(g.CacheID, key, SetPublishType, e)
				if _err != nil {
					LogErr("syncLocalCache Publish err:", err)
				}
			}
		})
	}
	// 如果缓存数据源获取到了数据,且数据没过期,则命中次数加1
	atomic.AddInt64(&HitStatisticsOut.HitOutStorageTotal, 1)
	if CacheDebug {
		LogDebugF("key:%-15s => [\u001B[33m hit out storage \u001B[0m]\n", key)
	}
	// 最后将本地数据源数据更新
	return g.local.Set(key, e)
}

func (g *MultiCache) syncOutCache(key string, ttlSecond int, fn LoadDataSourceFunc) (interface{}, error) {
	// 使用外部数据源获取数据次数+1
	atomic.AddInt64(&HitStatisticsOut.HitDataSourceTotal, 1)
	if CacheDebug {
		LogDebugF("key:%-15s => [\u001B[31m hit data source \u001B[0m]\n", key)
	}
	// 对当前分片上锁
	g.getShardsLock(key)
	defer g.releaseShardsLock(key)
	// 获取数据
	o, err := fn()
	if err != nil {
		return nil, err
	}
	// 如果数据为空则返回nil
	if o == nil {
		return nil, DataSourceLoadNil
	}
	// 存在对于数据,构造entry
	e := NewEntry(o, ttlSecond)
	// 将构造的entry存入本地缓存
	err = g.local.Set(key, e)
	if err != nil {
		return nil, err
	}
	// 将构造的entry同步到缓存数据源
	g.workerPool.SendJob(func() {
		_err := g.out.Set(key, e)
		if _err != nil {
			return
		}
		// 如果缓存数据源实现了发布订阅模型
		pubsub, ok := g.out.(PubSub)
		if ok && OutCachePubSub {
			// 还需要将对应的数据同步到整个集群
			_err = pubsub.Publish(g.CacheID, key, SetPublishType, e)
			if _err != nil {
				eS, _ := json.MarshalToString(e)
				LogErrF("syncOutCache key=%s,val=%s,err=%v\n", key, eS, err)
			}
		}
	})

	return o, err
}

// 对传入的数据key进行分片,缩小锁的返回,提高效率
func (g *MultiCache) getShardsLock(key string) {
	idx := g.hash.Sum64(key)
	g.shardsLock[idx&defaultShardsAndOpVal].Lock()
}

// 释放分片锁
func (g *MultiCache) releaseShardsLock(key string) {
	idx := g.hash.Sum64(key)
	g.shardsLock[idx&defaultShardsAndOpVal].Unlock()
}

// Set 存入数据
// key 存入数据对应key
// obj 存入数据对应值
// wait 是否同步等待数据存入 true则等待数据存入完成同步返回,false异步存入数据
func (g *MultiCache) Set(key string, obj interface{}, ttlSecond int, wait bool) (err error) {
	select {
	// 缓存关闭
	case <-g.isStop:
		return CacheClose
	default:
	}
	// 校验key不为空
	if key == "" {
		return CacheKeyEmpty
	}
	// 校验数据不为空
	if obj == nil {
		return CacheObjNil
	}
	// 设置过期时间,默认为5秒
	if ttlSecond <= 0 {
		ttlSecond = 5
	}
	return g.set(key, obj, ttlSecond, wait)
}

func (g *MultiCache) set(key string, obj interface{}, ttlSecond int, wait bool) (err error) {
	// 构造一个数据的entry
	v := NewEntry(obj, ttlSecond)
	if wait {
		// 同步存入数据
		return g.setInternal(key, v)
	}
	// 异步存入数据
	g.workerPool.SendJob(func() {
		_err := g.setInternal(key, v)
		if _err != nil {
			objS, _ := json.MarshalToString(v)
			LogErrF("setInternal key: %s,obj: %s ,err: %v", key, objS, err)
		}
	})

	return err
}

func (g *MultiCache) setInternal(key string, e *Entry) (err error) {
	// 存入本地数据源
	err = g.local.Set(key, e)
	if err != nil {
		return err
	}
	// 存入缓存数据源
	err = g.out.Set(key, e)
	if err != nil {
		return err
	}
	// 同步数据到发布订阅集群
	pubsub, ok := g.out.(PubSub)
	if ok && OutCachePubSub {
		return pubsub.Publish(g.CacheID, key, SetPublishType, e)
	}
	return nil
}

// Del 删除数据
// key 数据删除数据的key
// wait 是否同步删除
func (g *MultiCache) Del(key string, wait bool) (err error) {
	select {
	case <-g.isStop:
		// 缓存关闭
		return CacheClose
	default:
	}
	// 校验key不为空
	if key == "" {
		return CacheKeyEmpty
	}
	return g.del(key, wait)
}

func (g *MultiCache) del(key string, wait bool) (err error) {
	if wait {
		// 同步删除数据
		return g.delInternal(key)
	}
	// 异步删除数据
	g.workerPool.SendJob(func() {
		_err := g.delInternal(key)
		if _err != nil {
			LogErrF("delInternal key: %s,err: %v", key, err)
		}
	})

	return nil
}

func (g *MultiCache) delInternal(key string) (err error) {
	defer func() {
		// 如果外部数据以及发布订阅源数据删除成功
		if err == nil {
			// 删除本地数据源数据
			err = g.local.Del(key)
		}
	}()
	// 删除缓存数据源数据
	err = g.out.Del(key)
	if err != nil {
		return err
	}
	// 删除发布订阅集群数据
	pubsub, ok := g.out.(PubSub)
	if ok && OutCachePubSub {
		return pubsub.Publish(g.CacheID, key, DelPublishType, nil)
	}
	return err
}

func (g *MultiCache) subscribe() error {
	select {
	// 获得停止信号,停止当前缓存
	case <-g.isStop:
		return CacheClose
	default:
	}
	return g.subscribeInternal()
}

// subscribeHandle 负责处理所有发布订阅类型的数据
func (g *MultiCache) subscribeHandle() error {
	// 从数据channel中获取数据
	for ele := range g.channel {
		meta := *ele // for range pointer please to do
		select {
		case <-g.isStop:
			return OutStorageClose
		default:
		}
		if meta.CacheID == g.CacheID {
			continue
		}
		if meta.Key == "" {
			if CacheDebug {
				LogDebugF("subscribeHandle receive meta.Key is null: %+v\n", meta)
			}
			continue
		}
		if CacheDebug {
			metaDump, _ := json.MarshalToString(meta)
			LogDebugF("subscribeHandle receive meta: %v\n", metaDump)
		}
		// 数据的类型
		switch meta.Action {
		case DelPublishType:
			// 删除数据
			g.workerPool.SendJob(func() {
				if err := g.out.Del(meta.Key); err != nil {
					LogErrF("out del key=%s, err=%v\n", meta.Key, err)
				}
				if err := g.local.Del(meta.Key); err != nil {
					LogErrF("local del key=%s, err=%v\n", meta.Key, err)
				}
			})
		case SetPublishType:
			// 发布数据
			if meta.Data.Value == nil {
				if CacheDebug {
					LogDebugF("subscribeHandle receive meta.Data is nil: %+v\n", meta)
				}
				continue
			}
			g.workerPool.SendJob(func() {
				// 存入本地缓存
				if err := g.local.Set(meta.Key, meta.Data); err != nil {
					dataS, _ := json.MarshalToString(meta.Data)
					LogErrF("local set key=%s,val=%s, err=%v\n", meta.Key, dataS, err)
				}
				// 存入外部缓存
				if err := g.out.Set(meta.Key, meta.Data); err != nil {
					dataS, _ := json.MarshalToString(meta.Data)
					LogErrF("out set key=%s,val=%s, err=%v\n", meta.Key, dataS, err)
				}
			})
		default:
			continue
		}
	}
	return nil
}

func (g *MultiCache) subscribeInternal() error {
	// 获取缓存数据源
	pubsub, ok := g.out.(PubSub)
	if !ok {
		// 未实现发布订阅接口
		return CacheNotImplementPubSub
	}
	// 如果实现了发布订阅接口,将发布订阅数据的处理函数送入协程池
	g.workerPool.SendJob(wrapFuncErr(func() error {
		return g.subscribeHandle()
	}))
	// 传入对应的channel接收数据
	err := pubsub.Subscribe(g.channel)
	if err != nil {
		return err
	}

	return nil
}

// close 关闭缓存
func (g *MultiCache) close() {
	if g.isStop != nil {
		close(g.isStop)
	}
	// 关闭外部缓存
	if g.out != nil {
		g.out.Close()
	}
	// 关闭本地缓存
	if g.local != nil {
		g.local.Close()
	}
	// 关闭数据channel
	if g.channel != nil {
		close(g.channel)
	}
	// 关闭线程池
	if g.workerPool != nil {
		g.workerPool.Release()
	}
}

// Close 使用once确保所有线程只能关闭一次
func (g *MultiCache) Close() {
	g.stopLock.Do(g.close)
}

type Harsher interface {
	Sum64(string) uint64
}

type fnv64a struct{}

const (
	// offset64 FNVa offset basis. See https://en.wikipedia.org/wiki/Fowler–Noll–Vo_hash_function#FNV-1a_hash
	offset64 = 14695981039346656037
	// prime64 FNVa prime value. See https://en.wikipedia.org/wiki/Fowler–Noll–Vo_hash_function#FNV-1a_hash
	prime64 = 1099511628211
)

// Sum64 哈希函数,对key进行hash.
func (f fnv64a) Sum64(key string) uint64 {
	var hash uint64 = offset64
	for i := 0; i < len(key); i++ {
		hash ^= uint64(key[i])
		hash *= prime64
	}

	return hash
}
